#region Using References

using System;
using System.Collections.Generic;
using System.Collections.ObjectModel;
using System.Diagnostics.Contracts;
using System.Threading;
using System.Threading.Tasks;

#endregion

namespace Beaker.Services.Tasks
{
	internal abstract class TaskCollection<T>
		: ITaskCollection<T>
	{
		private readonly Collection<Task<T>> _tasks = new Collection<Task<T>>();

		/// <summary>
		/// Starts the given task and expects a result
		/// </summary>
		/// <typeparam name="T">The type of the result</typeparam>
		/// <param name="func">The function executing the task and returning the result</param>
		/// <returns>The task that will result at some point a result of type T</returns>
		public Task<T> Execute( Func<T> func )
		{
			Task<T> task = ExecuteCore( func );
			_tasks.Add( task );
			return task;
		}

		protected abstract Task<T> ExecuteCore( Func<T> func );

		/// <summary>
		/// Observes the results of all the tasks in the collection
		/// </summary>
		/// <typeparam name="T">The type of the result</typeparam>
		/// <param name="keepOrder">Set to true to keep the order of the results</param>
		/// <returns>An observable to observe the results asynchronously</returns>
		public IObservable<T> ObserveResults( bool keepOrder = false )
		{
			// we'll start a new task :)
			// that task will observe the other tasks and send the results to the observable
			var observable = new Observable( this );
			Task.Run(
				() =>
				{
					var awaitableTasks = new List<Task<T>>( _tasks );

					try
					{
						if( keepOrder )
						{
							throw new NotImplementedException( "keeping order not yet implmented" );
						}
						else
						{
							int totalTasks = awaitableTasks.Count;
							for( int i = 0; i < totalTasks; i++ )
							{
								// wait unit any of the tasks finishes
								int index = Task.WaitAny( awaitableTasks.ToArray() );

								// get the task end send result or exception to observable
								Task<T> task = awaitableTasks[index];
								if( task.IsFaulted )
									observable.RaiseException( task.Exception );
								else
									observable.SendResult( task.Result );

								// remove from the list of tasks we can still await
								awaitableTasks.RemoveAt( index );
							}
						}

						// all tasks are done.
						observable.SendCompleted();
					}
					catch( Exception x )
					{
						observable.RaiseException( x );
					}
				}
				);

			return observable;
		}

		private class Observable
			: IObservable<T>
		{
			private readonly TaskCollection<T> _collection;

			public Observable( TaskCollection<T> collection )
			{
				Contract.Requires<ArgumentNullException>( collection != null );
				_collection = collection;
			}

			private readonly Collection<IObserver<T>> _observers =
				new Collection<IObserver<T>>();

			// TODO: Maybe upgrade to r/w lock because there are many reads, and almost no writes
			private readonly object _observersLock = new object();

			/// <summary>
			/// Notifies the provider that an observer is to receive notifications.
			/// </summary>
			/// <returns>
			/// A reference to an interface that allows observers to stop receiving notifications before the provider has finished sending them.
			/// </returns>
			/// <param name="observer">The object that is to receive notifications.</param>
			public IDisposable Subscribe( IObserver<T> observer )
			{
				//Contract.Requires<ArgumentNullException>( observer != null );

				// add observer
				lock( _observersLock )
				{
					_observers.Add( observer );
					Monitor.PulseAll( _observersLock );
				}

				// unsubscribe on dispose
				return new ActionOnDispose(
					() => Unsubscribe( observer )
					);
			}

			private void Unsubscribe( IObserver<T> observer )
			{
				Contract.Requires<ArgumentNullException>( observer != null );

				// remove from list of subscribers
				lock( _observersLock )
				{
					_observers.Remove( observer );
				}
			}

			public void SendResult( T update )
			{
				lock( _observersLock )
				{
					if( _observers.Count == 0 )
						Monitor.Wait( _observersLock );

					foreach( IObserver<T> observer in _observers )
						observer.OnNext( update );
				}
			}

			public void RaiseException( Exception x )
			{
				lock( _observersLock )
				{
					if( _observers.Count == 0 )
						Monitor.Wait( _observersLock );

					foreach( IObserver<T> observer in _observers )
						observer.OnError( x );
				}
			}

			public void SendCompleted()
			{
				lock( _observersLock )
				{
					if( _observers.Count == 0 )
						Monitor.Wait( _observersLock );

					foreach( IObserver<T> observer in _observers )
						observer.OnCompleted();
				}
			}
		}
	}
}